package com.ydl.learning.flink.demo

import java.time.ZoneId
import java.util.Properties
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

/**
 *
 *
 * @author ydl
 * @since 2020/10/10
 */
trait TableUtils {
  val dataDemo = List((1, "a", 1.2), (2, "b", 2.1), (1, "a", 1.1))
  val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
  val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  streamEnv.disableOperatorChaining()

  //cmd nc -lk 9999
  def getSocketTextStream: (StreamExecutionEnvironment, DataStream[String]) = {
    (streamEnv, streamEnv.socketTextStream("localhost", 9999, '\n'))
  }

  def readFile: DataSet[String] = {
    env.readTextFile("/Volumes/work/code/study/learning-flink/pom.xml")
  }

  val kafkaProperties = new Properties()
  kafkaProperties.put("bootstrap.servers", "10.50.168.216:9092,10.50.168.148:9092,10.50.168.242:9092")
  kafkaProperties.put("group.id", "flink_test")
  kafkaProperties.put("auto.offset.reset", "latest")
  kafkaProperties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
  kafkaProperties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
  val consumer: FlinkKafkaConsumer[String] = new FlinkKafkaConsumer[String]("test", new SimpleStringSchema(), kafkaProperties)

  //blink配置,useBlinkPlanner明确设置应用里使用blink计划器，或者useOldPlanner设置使用Flink计划器
  //当项目里只有一种计划器的jar包时，可以使用useAnyPlanner来设置计划器
  //注意流和批的Mode不同，流用inStreamingMode，批用inBatchMode
  val bsSettings: EnvironmentSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
  val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(streamEnv, bsSettings)
  // set configuration early
  //tableEnv.getConfig.setLocalTimeZone(ZoneId.of("Europe/Berlin"))


  //val bbSettings: EnvironmentSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
  //val bbTableEnv: TableEnvironment = TableEnvironment.create(bbSettings)

}
